package groupOne.app.DWD.db;


import groupOne.app.BaseAppSql;
import groupOne.common.Constant;
import groupOne.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdOrderDetailXiang extends BaseAppSql {
    public static void main(String[] args) {
        new DwdOrderDetailXiang().init(
                3001,
                2,
                "Dwd_Order_Detail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"Dwd_Order_Detail");
        Table result = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['course_id'] course_id, " +
                "data['course_name'] course_name, " +
                "data['order_id'] order_id, " +
                "data['user_id'] user_id, " +
                "data['origin_amount'] origin_amount, " +
                "data['coupon_reduce'] coupon_reduce, " +
                "data['final_amount'] final_amount, " +
                "data['create_time'] create_time, " +
                "data['update_time'] update_time," +
                "ts " +
                "from ods_db " +
                "where `database` = 'edu' " +
                "and `table` = 'order_detail' " +
                "and `type` = 'insert' ");
        tEnv.executeSql("create table dwd_trade_order_detail(" +
                "id string, " +
                "course_id string, " +
                "course_name string, " +
                "order_id string, " +
                "user_id string, " +
                "origin_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "create_time string, " +
                "update_time string," +
                "ts bigint " +
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL)
        );

        result.executeInsert("dwd_trade_order_detail");

    }
}
